Amazon Transcribe のパイプラインで中国語学習を加速する

哈喽大家好、コンサルティング部の西野です。

Amazon Transcribe を利用して、自己学習のための中国語文字起こしパイプラインを作ってみました。

解決したかったこと

筆者は趣味として日常的に中国語のコンテンツに触れています。

主に観ているものは映画・ドラマ・中国人YouTuberの動画・抖音 (中国版 Tiktok) などであり、これらのコンテンツのほとんどには普通话 (中国における標準語) の字幕がついています。字幕のおかげもあって、文字通りの意味を取るだけなら特に困ることがない状態です。

このまま単なる趣味に据え置くなら今のままでも良いのですが、最近は「せっかく勉強したのにもったいない！もう少し脳に負担がかかるコンテンツを使ってレベル上げをしたい！」と思うようになりました。

そこで目をつけたのが NHK WORLD RADIO JAPAN News の Chinese News です。NHK WORLD RADIO JAPAN News は世界各国の言語で主に日本のニュースを発信している Podcast です。

「これを使えば中国語も勉強できるしニュースにも触れられる。一石二鳥だ」と軽い気持ちで聴いてみたのですが、正直言ってすこし難しい。どのようなテーマの話をしているかは大体わかるものの、ディティールについては聞き逃すことがままあります。

もしニュースのスクリプトがあれば難易度としてちょうどいいと感じました。ただし、残念ながら、NHK WORLD RADIO JAPAN News は公式としてニュースのスクリプトを公開していないようでした。

やろうと思ったこと

スクリプトがないなら自分で作ってしまえばいいのです。 要件はざっくり以下のとおりにしました。

• Amazon Transcribe を使って文字起こしをする
• 文字起こし結果を取得するまでの手間をなるべく少なくする
  • Amazon Transcribe のジョブを毎回マネジメントコンソールでぽちぽち作って……ということをしたくない。
• いつも使っているツール (今回の場合は Slack ) に結果を通知したい

構成図

こうしてできあがったのが下図の仕組みです。

① インプット用バケットに Podcast の mp3 ファイルをアップロード
② S3 イベントでジョブ実行 Lambda 関数をトリガー
③ ジョブ実行 Lambda 関数が Transcribe のジョブを開始
④ ジョブ実行 Lambda 関数が Slack チャンネルに Transcribe ジョブの開始を通知 (Incoming Webhook を利用)
⑤ Transcribe ジョブの実行結果（ json ファイル）をアウトプット用バケットに保存
⑥ S3 イベントで URL 生成 Lambda 関数をトリガー
⑦ URL 生成 Lambda 関数が json ファイルに対する署名付き URL を発行し、Slack チャンネルに通知 (Incoming Webhook を利用)

使ってみた

① におけるファイルのアップロードはどのように行っても構いませんが、今回は CLI から実施します。

下記のポリシーのみを持つ IAM ユーザーを作成し、クレデンシャルを取得します。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:PutObject"
            ],
            "Resource": [
                "arn:aws:s3:::<インプット用バケット名>/*"
            ]
        }
    ]
}

当該 IAM ユーザーのクレデンシャルで名前付きプロファイルを作成します。

$ aws configure --profile transcribe_zh-cn
AWS Access Key ID [None]: AKIAXXXXXXXXXXXXXXXX
AWS Secret Access Key [None]: 2ajnqsXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
Default region name [None]: ap-northeast-1
Default output format [None]:

下記の (雑な) スクリプトを使って S3 に mp3 ファイルをアップロードします。今回は Python + Boto3 を使用しました。シェルスクリプト (AWS CLI) などでも良いと思います。

#!/usr/bin/env python
import sys
import boto3

profile_name = "transcribe_zh-cn"
bucket_name = "<インプット用バケット名>"
file_name = sys.argv[1]

if __name__ == '__main__':
    print("{} をアップロードします。".format(file_name))
    session = boto3.Session(profile_name=profile_name)
    s3 = session.resource('s3')
    bucket = s3.Bucket(bucket_name)
    bucket.upload_file(file_name, file_name)

$ ./upload.py 20201022_nhk_radio_chinese.mp3
20201022_nhk_radio_chinese.mp3 をアップロードします。

ファイルのアップロードが済むとすぐに Slack に通知が飛んできます。あわせて Transcribe のジョブも実行されています。裏で動いている Lambda 関数の詳細については後述します。

ジョブが完了するとダウンロードリンク ( S3 署名付き URL ) が通知されます。

リンクをクリックすると json ファイルをダウンロードできます。

求めていたニュースのスクリプトを入手できました。
音声を聴きながらスクリプトを眺めてみたところ、ラジオニュース番組という性質からか、かなり高い精度で文字起こししてくれていました。ところどころ間違いが見られるものの、当初の目的は十分達成できそうです。

各リソースの説明

インプット用バケット

文字起こし完了後は mp3 ファイルを残しておく必要がないため、S3 ライフサイクルルールを使用してオブジェクトが1日で削除されるように設定しています。

$ aws s3api get-bucket-lifecycle-configuration --bucket <インプット用バケット名>
{
    "Rules": [
        {
            "Expiration": {
                "Days": 1
            },
            "ID": "delete-old-mp3",
            "Filter": {
                "Prefix": ""
            },
            "Status": "Enabled"
        }
    ]
}

また、mp3 ファイル (オブジェクトのサフィックスで指定) がアップロードされた場合のみ後述のジョブ実行 Lambda をトリガーするように設定してあります。

$ aws s3api get-bucket-notification-configuration --bucket <インプット用バケット名>
{
    "LambdaFunctionConfigurations": [
        {
            "Id": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
            "LambdaFunctionArn": "arn:aws:lambda:ap-northeast-1:XXXXXXXXXXXX:function:start_transcription_job",
            "Events": [
                "s3:ObjectCreated:*"
            ],
            "Filter": {
                "Key": {
                    "FilterRules": [
                        {
                            "Name": "Suffix",
                            "Value": ".mp3"
                        }
                    ]
                }
            }
        }
    ]
}

ジョブ実行 Lambda (start_transcription_job)

Transcribe ジョブの実行と Slack への通知を実行している Lambda 関数です。

import datetime
import json
import urllib.request
import boto3

s3 = boto3.client('s3')
transcribe = boto3.client('transcribe')

webhook_url = "<https://hooks.slack.com/services/XXXXXXXXX/XXXXXXXXXX/XXXXXXXXXXXXXXXXXXXXXXXX>"
username = "中国語文字おこしちゃん"
channel = "#times-nishino-wataru"
output_bucket_name = "<アウトプット用バケット名>"

def lambda_handler(event, context):
    bucket_name = event["Records"][0]["s3"]["bucket"]["name"]
    object_key = event["Records"][0]["s3"]["object"]["key"]
    
    file_uploaded_message = "{} がアップロードされました。".format(object_key)
    send_to_slack(webhook_url, username, channel, file_uploaded_message)

    response = start_transcription_job(bucket_name, object_key)
    send_to_slack(webhook_url, username, channel, response)
    

    
def send_to_slack(webhook_url, username, channel, message):
    send_data = {
        "username": username,
        "text": message,
        "channel": channel
    }

    send_text = ("payload=" + json.dumps(send_data)).encode('utf-8')

    request = urllib.request.Request(
        webhook_url,
        data=send_text,
        method="POST"
    )
    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode('utf-8')

    return response_body

def start_transcription_job(bucket_name, object_key):
    try:
        response = transcribe.start_transcription_job(
            TranscriptionJobName = object_key.split(".")[0],
            LanguageCode='zh-CN',
            Media={
                'MediaFileUri': '<https://s3.ap-northeast-1.amazonaws.com/>' + bucket_name + '/' + object_key
            },
            OutputBucketName= output_bucket_name
        )
        print(response)
        return "文字起こしのジョブを開始しました。"
    except Exception as e:
        print(e)
        return "文字起こしのジョブを開始できませんでした。"

アウトプット用バケット

インプット用バケットとは別のバケットを作成しています。

また、Transcribe ジョブのアウトプットである json ファイル (オブジェクトのサフィックスで指定) がアップロードされた場合のみ後述の URL 生成 Lambda をトリガーするように設定してあります。

$ aws s3api get-bucket-notification-configuration --bucket <アウトプット用バケット>

{
    "LambdaFunctionConfigurations": [
        {
            "Id": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
            "LambdaFunctionArn": "arn:aws:lambda:ap-northeast-1:XXXXXXXXXXXX:function:send_s3url_to_slack",
            "Events": [
                "s3:ObjectCreated:*"
            ],
            "Filter": {
                "Key": {
                    "FilterRules": [
                        {
                            "Name": "Suffix",
                            "Value": ".json"
                        }
                    ]
                }
            }
        }
    ]
}

URL 生成 Lambda (send_s3url_to_slack)

アウトプットである json ファイルのダウンロード用署名付き URL の生成と Slack への通知を実行している Lambda 関数です。

import json
import urllib.request
from urllib.parse import quote
import boto3

s3 = boto3.client('s3')

webhook_url = "<https://hooks.slack.com/services/XXXXXXXXX/XXXXXXXXXX/XXXXXXXXXXXXXXXXXXXXXXXX>"
username = "中国語文字おこしちゃん"
channel = "#times-nishino-wataru"

def lambda_handler(event, context):
    bucket_name = event["Records"][0]["s3"]["bucket"]["name"]
    object_key = event["Records"][0]["s3"]["object"]["key"]
    s3_url = generate_presigned_url(bucket_name, object_key)
    message = "文字起こしが完了しました。\\nダウンロードリンクはこちらです。\\n<{}|リンク>".format(s3_url)
    response = send_to_slack(webhook_url, username, channel, message)
    return response

def generate_presigned_url(bucket_name, object_key):
    presigned_url = s3.generate_presigned_url(
        ClientMethod = 'get_object',
        Params = {'Bucket' : bucket_name, 'Key' : object_key},
        ExpiresIn = 3600,
        HttpMethod = 'GET'
        )
    quoted_presigned_url = quote(presigned_url)
    return quoted_presigned_url

def send_to_slack(webhook_url, username, channel, message):
    send_data = {
        "username": username,
        "text": message,
        "channel": channel
    }

    send_text = ("payload=" + json.dumps(send_data)).encode('utf-8')

    request = urllib.request.Request(
        webhook_url,
        data=send_text,
        method="POST"
    )
    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode('utf-8')

    return response_body

参考ブログなど

本ブログの仕組みを構築するにあたっては下記のブログやドキュメントを参考にいたしました。

AWS Lambdaを利用してpythonでslack通知を実装する - ギークなエンジニアを目指す男

API Gateway + LambdaアプリケーションでS3に署名付きURLを生成しリダイレクトする | Developers.IO

JAWS-UG 初心者支部#24 サーバレスハンズオン勉強会にリモートで参加しました！ #jawsug_bgnr | Developers.IO

SlackのWeb APIでURLを含むメッセージがポストできなかった話 - Qiita

StartTranscriptionJob

終わりに

このブログがほんの少しでも世界を良くできれば嬉しいです。
コンサルティング部の西野 (@xiyegen) がお送りしました。